#pragma once
#include <sys/socket.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <iostream>
#include "log.hpp"
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include "buffer.hpp"
#include "any.hpp"
#define MAX_LISTEN 5
#define MAX_EPOLL_SIZE 1024 // 无意义大于0就行
#define MAX_TIME_SCALE 60   // 时间轮最大刻度
class Socket
{
public:
    Socket() : _socketfd(-1) {}
    Socket(int fd) : _socketfd(fd) {}
    ~Socket() { Close(); }

    int Fd() { return _socketfd; }
    bool CreatSocket()
    {
        _socketfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_socketfd < 0)
        {
            lg.logmessage(Error, "CreatSocket failure %s", strerror(errno));
            return false;
        }
        // std::cout << _socketfd << std::endl;
        lg.logmessage(Info, "CreatSocket success");
        return true;
    }
    bool Bind(const std::string &ip, uint16_t port)
    {
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        addr.sin_port = htons(port);
        socklen_t len = sizeof(struct sockaddr_in);
        if (bind(_socketfd, (struct sockaddr *)&addr, len) < 0)
        {
            lg.logmessage(Error, "bind failure %s", strerror(errno));
            return false;
        }
        lg.logmessage(Info, "bind sucess");
        return true;
    }
    bool Listen(int backlog = MAX_LISTEN)
    {
        if (listen(_socketfd, MAX_LISTEN) < 0)
        {
            lg.logmessage(Error, "listen failure %s", strerror(errno));
            return false;
        }
        lg.logmessage(Info, "listen sucess");
        return true;
    }
    int Accept()
    {
        int fd = accept(_socketfd, nullptr, nullptr);
        if (fd < 0)
        {
            lg.logmessage(Error, "accept failure %s", strerror(errno));
            return -1;
        }
        lg.logmessage(Info, "accept sucess");
        return fd;
    }
    ssize_t Recive(void *buf, size_t n, int flag = 0)
    {
        ssize_t ret;
        ret = recv(_socketfd, buf, n, flag);
        if (ret < 0)
        {
            /// EAGAIN 无数据
            /// EINTR(enter interupt)被中断打断
            if (errno == EAGAIN || errno == EINTR)
            {
                lg.logmessage(Warning, "recive warning");
                return 0;
            }
            lg.logmessage(Error, "recive failure %s", strerror(errno));
            return -1;
        }
        lg.logmessage(Info, "recive suess");
        return ret;
    }

    ssize_t NoneBlockRecv(void *buf, size_t n)
    {
        return Recive(buf, n, MSG_DONTWAIT);
    }
    ssize_t Send(const void *buf, size_t n, int flag = 0)
    {
        int ret;
        ret = send(_socketfd, buf, n, flag);
        if (ret < 0)
        {
            lg.logmessage(Error, "send failure %s", strerror(errno));
            return -1;
        }
        else
        {
            lg.logmessage(Info, "send suess");
        }
        return ret;
    }

    ssize_t NoneBlockSend(const void *buf, size_t n)
    {
        return Send(buf, n, MSG_DONTWAIT);
    }

    bool Connect(const std::string &ip, uint16_t port)
    {
        sockaddr_in server;
        server.sin_family = AF_INET;
        server.sin_addr.s_addr = inet_addr(ip.c_str());
        server.sin_port = htons(port);
        if (connect(_socketfd, (sockaddr *)&server, sizeof(sockaddr_in)))
        {
            lg.logmessage(Error, "connect error %s", strerror(errno));
            return false;
        }
        lg.logmessage(Info, "connect succes");
        return true;
    }

    bool CreatServer(uint16_t port, const std::string &ip = "0.0.0.0", bool noneBlock = false)
    {

        if (CreatSocket())
        {
            ResuseAddrAndPort();
            if (Bind(ip, port))
            {

                if (noneBlock)
                    NoneBlockSocket();
                if (Listen())
                {

                    return true;
                }
            }
        }
        return false;
    }

    bool CreatClient(std::string ip, uint16_t port)
    {
        if (CreatSocket())
        {
            if (Connect(ip, port))
                return true;
        }
        return false;
    }

    /// @brief 端口地址复用
    bool ResuseAddrAndPort()
    {
        int ret = 0;
        int opt = 1;
        ret = setsockopt(_socketfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(int));
        if (ret < 0)
        {
            lg.logmessage(Error, "setsockopt error %s", strerror(errno));
            return false;
        }
        opt = 1;
        ret = setsockopt(_socketfd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(int));
        if (ret < 0)
        {
            lg.logmessage(Error, "setsockopt error %s", strerror(errno));
            return false;
        }
        lg.logmessage(Info, "setsockopt  success");
        return true;
    }

    void NoneBlockSocket()
    {
        int flag = fcntl(_socketfd, F_GETFL, 0);
        fcntl(_socketfd, F_SETFL, flag | O_NONBLOCK);
    }

    void Close()
    {
        if (_socketfd != -1)
        {
            std::cout << "close " << _socketfd << std::endl;
            close(_socketfd);
            _socketfd = -1;
            return;
        }
        std::cout << __LINE__ << " have closed " << _socketfd << std::endl;
        _socketfd = -1;
    }

private:
    int _socketfd;
};

class Poller;
class EventLoop;

using TaskFunc = std::function<void(void)>;
/// @brief 管理事件
class Channel
{
private:
    int _fd;
    EventLoop *_loop;
    uint32_t _event;         // 当前监控的事件
    uint16_t _current_event; // 当前正在发生的（你发生的事件只有可能是监控中）
    using EventCallback = std::function<void(void)>;
    EventCallback _read_callback;
    EventCallback _write_callback;
    EventCallback _close_callback;
    EventCallback _error_callback;
    EventCallback _event_callback;

public:
    Channel(int fd, EventLoop *loop) : _fd(fd), _loop(loop), _event(0), _current_event(0)
    {
        _event = _event | EPOLLHUP;
        _event = _event | EPOLLPRI;
        _event = _event | EPOLLERR;
        _event = _event | EPOLLHUP;
    }
    ~Channel()
    {
        // if (_fd > 0)
        //     close(_fd);
        _loop = nullptr;
        _event = 0;
        _current_event = 0;
    }

#pragma region 设置回调
    void SetReadCallback(const EventCallback &cb) { _read_callback = cb; }
    void SetWriteCallback(const EventCallback &cb) { _write_callback = cb; }
    void SetCloseCallback(const EventCallback &cb) { _close_callback = cb; }
    void SetErrorCallback(const EventCallback &cb) { _error_callback = cb; }
    void SetEventCallback(const EventCallback &cb) { _event_callback = cb; }
    void SetCurrentEvent(uint32_t event) { _current_event = event; };
#pragma endregion
    int Fd() { return _fd; }
    uint32_t Event() { return _event; }
    bool isReadable() { return _event & EPOLLIN; }   // EPOLLIN可读
    bool isWriteable() { return _event & EPOLLOUT; } // EPOLLOUT可写 不可以在可写的基础上再次修改为可写epoll特性
    void EnableRead()
    {
        _event |= EPOLLIN;
        Update();
    }
    void EnableWrite()
    {
        _event |= EPOLLOUT;
        Update();
    }
    void DisableRead()
    {
        _event &= ~EPOLLIN;
        Update();
    }
    void DisableWrite()
    {
        _event &= ~EPOLLOUT;
        Update();
    }
    void DisableAll()
    {
        _event = 0;
        Update();
    }
    void Remove();
    void Update();
    void HandleEvent() // 处理事件
    {
        if ((_current_event & EPOLLIN) || (_current_event & EPOLLHUP) || (_current_event & EPOLLPRI)) // 读事件 对方关闭 优先带外数据
        {

            if (_read_callback)
                _read_callback();
        }
        // std::cout << __LINE__ << " : " << Fd() << _current_event << std::endl;
        if (_current_event & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        else if (_current_event & EPOLLHUP)
        {
            lg.logmessage(Debug, "EPOLLHUP");
            if (_close_callback)
                _close_callback();
        }
        else if (_current_event & EPOLLERR)
        {
            lg.logmessage(Debug, "EPOLLERR");

            if (_error_callback)
                _error_callback();
        }

        if (_event_callback) // 刷新活跃度使业务处理时间不算在活跃度内
            _event_callback();
    }
};

/// @brief 管理连接? 需要操控的一直是poller里的channel
class Poller
{

private:
    void Update(Channel *channel, int op)
    {
        epoll_event event;
        event.data.fd = channel->Fd();
        event.events = channel->Event();

        if (epoll_ctl(_epfd, op, channel->Fd(), &event) < 0)
        {
            lg.logmessage(Error, "%d:%d update %d error  %s", _epfd, op, channel->Fd(), strerror(errno));
            abort();
        }
    }

    bool HasChannel(Channel *channel)
    {
        auto item = _channels.find(channel->Fd());
        return item != _channels.end();
    }
    void CreatPool()
    {
        _epfd = epoll_create(MAX_EPOLL_SIZE);
    }

public:
    Poller(int timeout = -1) : _timeout(timeout)
    {
        CreatPool();
    }

    /// @brief 如果没有加入管理则直接加入管理 如果加入过则改变
    /// @param channel
    void UpdateEvent(Channel *channel)
    {

        if (HasChannel(channel))
        {
            Update(channel, EPOLL_CTL_MOD); // 有就改变
        }
        else
        {
            Update(channel, EPOLL_CTL_ADD); // 没有就添加
            _channels.insert(std::make_pair(channel->Fd(), channel));
            lg.logmessage(Debug, "add a new channle %d", channel->Fd());
        }
    }

    void RemoveEvent(Channel *channel)
    {
        if (HasChannel(channel))
        {
            Update(channel, EPOLL_CTL_DEL);
            _channels.erase(channel->Fd());
        }
        else
            lg.logmessage(Warning, "remove event not find fd %d", channel->Fd()); // 没找到这个fd
    }

    int Poll(std::vector<Channel *> *active)
    {
        epoll_event events[MAX_EPOLL_SIZE];
        int num = 0;

        num = epoll_wait(_epfd, events, MAX_EPOLL_SIZE, -1);
        // lg.logmessage(Debug, " in poll %d", num);
        if (num <= 0)
        {
            if (errno = EINTR)
                return 0;
            lg.logmessage(Error, "poll epoll_wait error %s", strerror(errno));
            // abort();
        }
        for (int i = 0; i < num; i++)
        {
            int fd = events[i].data.fd;
            if (_channels.find(fd) == _channels.end())
            {
                lg.logmessage(Error, "poll not find fd %d", events[i].data.fd);
                abort();
            }
            _channels[fd]->SetCurrentEvent(events[i].events);
            active->push_back(_channels[fd]);
        }
        return num;
    }

private:
    int _epfd;
    std::unordered_map<int, Channel *> _channels;
    int _timeout;
};

class TimerTask
{
    using ReleaseFunc = std::function<void(void)>;

public:
    TimerTask(uint64_t id, uint32_t timeout, const TaskFunc &task) : _id(id), _timeout(timeout), _task(task), _cancle(false)
    {
    }
    ~TimerTask()
    {
        // std::cout << __LINE__ << " ooo" <<_cancle<< std::endl;
        if (!_cancle && _task)
            _task();

        _release();
    }

    inline uint64_t TimeOut()
    {
        return _timeout;
    }

    void SetRelease(const ReleaseFunc &fuc) { _release = fuc; }
    void Cancle()
    {
        _cancle = true;
        // std::cout << "cancle" << _id << std::endl;
    }

private:
    uint64_t _id;      // 无符号64位整数 id可能会比较多所以要用这个
    uint32_t _timeout; // 设置的超时时间指延时几秒后再执行任务
    TaskFunc _task;    // 回调函数即任务
    ReleaseFunc _release;
    bool _cancle; // 标志者任务是否有效
};

class TimerWheel
{
    using SharedTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;

private:
    void RemoveTask(uint64_t id)
    {
        lg.logmessage(Info, "remove task is %d", id);
        if (_time_task.find(id) == _time_task.end())
            return;
        _time_task.erase(id);
    }
    int static CreatTimerFd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            perror("timerfd_create");
        }
        itimerspec t;
        t.it_value.tv_sec = 1;
        t.it_value.tv_nsec = 0;
        t.it_interval.tv_sec = 1;
        t.it_interval.tv_nsec = 0;
        if (timerfd_settime(timerfd, 0, &t, NULL) < 0)
        {
            // perror("timerfd_settime");
            lg.logmessage(Error, "timerfd_settime");
            abort();
        }
        return timerfd;
    }
    int ReadTimerFd()
    {
        uint64_t times;
        int ret = read(_timer_fd, &times, sizeof(uint64_t));
        if (ret < 0)
        {
            lg.logmessage(Error, "read timerfd failure %s", strerror(errno));
            abort();
        }
        return ret; // 因为在一个线程内事件都是排队的导致到执行此函数时已经过去很多s了
    }
    void Run()
    {
        _wheel[_tick].clear();             // 执行指定位置的任务
        _tick = (_tick + 1) % _time_scale; // 要先执行(当timeout=0时)
    }

    /// @brief timefd到时间会唤醒读事件然后清空timefd缓冲区执行任务
    void OnTime()
    {
        int times = ReadTimerFd();
        for (int i = 0; i < times; i++)
        {

            Run();
        }
    }

    void Add(uint64_t id, uint32_t timeout, const TaskFunc &task)
    {
        if (_time_task.find(id) != _time_task.end()) // 没有这个任务再往下走
        {
            return;
        }
        SharedTask st(new TimerTask(id, timeout, task));
        st->SetRelease(std::bind(&TimerWheel::RemoveTask, this, id)); // 由于要传递参数所以使用bind
        _wheel[(_tick + timeout) % _time_scale].push_back(st);        // 这里不会直接引用归零因为vector还拷贝赋值了一个
        _time_task[id] = WeakTask(st);
        _time_task.find(id);
    }

    /// @brief 刷新延时时间
    /// @param id
    void TimeRefresh(uint64_t id)
    {
        if (_time_task.find(id) == _time_task.end())
            return;
        SharedTask temp = _time_task[id].lock();
        _wheel[(_tick + temp->TimeOut()) % _time_scale].push_back(temp);
    }

    void CancleTask(uint64_t id)
    {
        if (_time_task.find(id) == _time_task.end())
            return;
        SharedTask st = _time_task[id].lock(); // weak_ptr.lock() 会把weak_ptr提升为一个 shared_ptr 对象，当 引用计数为0 ，那么 shared_ptr 为空。
        if (st)
        {
            // std::cout<<"???"<<std::endl;
            st->Cancle(); // 可能它已经处在析构中了
        }
    }

public:
    /// @brief
    /// @param timeScale 这个时间轮的最大刻度
    /// @param tick 时间轮的指针
    TimerWheel(EventLoop *loop, uint64_t timeScale = MAX_TIME_SCALE, uint64_t tick = 0) : _loop(loop),
                                                                                          _wheel(timeScale),
                                                                                          _tick(tick),
                                                                                          _time_scale(timeScale),
                                                                                          _timer_fd(CreatTimerFd()),
                                                                                          _channel_ptr(new Channel(_timer_fd, loop))
    {
        _channel_ptr->SetReadCallback(std::bind(&TimerWheel::OnTime, this));
        _channel_ptr->EnableRead();
    }
    void AddInLoop(uint64_t id, uint32_t timeout, const TaskFunc &task);
    void TimeRefreshInLoop(uint64_t id);
    void CancleTaskInLoop(uint64_t id);
    bool HasTime(uint64_t id)
    {
        if (_time_task.find(id) == _time_task.end())
            return false;
        return true;
    }

private:
    // 时间轮里面是个二维数组结构类似哈希桶 //_timr_task与——wheel的顺序不能变应为_wheel里的析构是对_time_task做操作所以_wheel要早与_time_task销毁
    std::unordered_map<uint64_t, WeakTask> _time_task; // 将每个任务存进去（方便进行操作后续涉及到刷新延时任务需要复制任务此时方便通过id找到它）
    std::vector<std::vector<SharedTask>> _wheel;
    uint64_t _time_scale;
    uint64_t _tick;
    int _timer_fd;                         // 定时器描述符
    EventLoop *_loop;                      // 时间监控只要你有事件要执行都要交给它
    std::unique_ptr<Channel> _channel_ptr; // 而_loop执行的对象即事件管理者channel
};

/// @brief 为了解决事件处理时线程冲突同时将事件管理起来逻辑上比较直观 time_whell顺带添加的额外超时模块
/// 每个eventloop 都有一个enent_channel 和timer_channel
class EventLoop
{
private:
    using Func = std::function<void(void)>;
    Poller _poller;             // 事件监控
    std::thread::id _thread_id; // 判断是否在这个线程？
    int _event_fd;              // event_id唤醒io事件监控有可能的阻塞？
    std::vector<Func> _tasks;
    std::mutex _mutex; // 这把锁的是对其管理的下面对_task的的操作因为只有这个是需要防范线程抢占的
    std::unique_ptr<Channel> _event_channel;
    TimerWheel _timer_whell;
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            lg.logmessage(Error, "efd creat error");
            abort();
        }
        return efd;
    }
    void ReadEventFd()
    {
        uint64_t res;
        int ret = read(_event_fd, &res, sizeof(uint64_t));
        if (ret < 0)
        {
            if (errno == EAGAIN || errno == EINTR) // 无数据可读 中断
            {
                return;
            }
            lg.logmessage(Error, "read eventfd failure");
            abort();
        }
    }
    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(uint64_t));
        if (ret < 0)
        {
            lg.logmessage(Error, "write eventfd failure");
            abort();
        }
    }

public:
    void RunAllTasks()
    {
        std::vector<Func> tasks; // 为了不影响对_tasks的插入
        {
            std::unique_lock<std::mutex> unique_mutex(_mutex);
            _tasks.swap(tasks);
        }
        for (auto item : tasks)
        {
            item();
        }
    }

public:
    EventLoop() : _thread_id(std::this_thread::get_id()),
                  _event_fd(CreateEventFd()),
                  _event_channel(new Channel(_event_fd, this)),
                  _timer_whell(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd, this));
        _event_channel->EnableRead();
    }

    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    void AssertInLoop()
    {
        assert(IsInLoop());
    }
    void RunInLoop(const Func &fc)
    {
        if (IsInLoop())
        {
            // std::cout<<"here";
            return fc();
        }
        QueueInLOOP(fc);
    }

    void QueueInLOOP(const Func &fc)
    {
        {
            
            std::unique_lock<std::mutex> unique_mutex(_mutex);
            _tasks.push_back(fc);
        }
        // 给eventfd写入一个数据，eventfd就会触发读事件
        WeakUpEventFd();
    }
    void UpdateEvent(Channel *channel)
    {
        _poller.UpdateEvent(channel);
    }
    void RemoveEvent(Channel *channel)
    {
        _poller.RemoveEvent(channel);
    }
    void Start()
    {
        while (1)
        {
            std::vector<Channel *> active;
            _poller.Poll(&active);
            for (auto item : active)
            {
                item->HandleEvent();
            }
            RunAllTasks(); // 当有人调用 QueueInLOOP即有人在不是本线程调用runinloop则将事件其插入队列并触发eventfd的读事件进而执行任务
        }
    }

    void TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task) { return _timer_whell.AddInLoop(id, timeout, task); }
    void TimerRefresh(uint64_t id) { return _timer_whell.TimeRefreshInLoop(id); }
    void TimerChannel(uint64_t id) { return _timer_whell.CancleTaskInLoop(id); }
    bool HasTimer(uint64_t id) { return _timer_whell.HasTime(id); }
};

class LoopThread
{

private:
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop *_loop;
    std::thread _thread; // eventloop对应的线程

    void ThreadEntry()
    {
        EventLoop loop; // 处在accept线程之外的loop其中管理的connection将争抢_task的资源
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all(); // 唤醒阻塞的线程？
        }
        _loop->Start();
    }

public:
    LoopThread() : _loop(nullptr), _thread(std::thread(&LoopThread::ThreadEntry, this))
    {
    }
    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]
                       { return _loop != nullptr; }); // 这个return return 为false才阻塞
            loop = _loop;                             //_loop不为空
        }
        return loop;
    }
};

class LoopThreadPool
{
private:
    uint64_t _next_idx;
    int _thread_count;
    EventLoop *_base_loop;
    std::vector<LoopThread *> _threads;
    std::vector<EventLoop *> _loops;

public:
    LoopThreadPool(EventLoop *base_loop, int thread_count = 0) : _next_idx(0), _thread_count(thread_count), _base_loop(base_loop) {}
    void SetThreadCount(int count) { _thread_count = count; }
    void Creat()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for (int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop(); // 如何得知线程就绪
            }
        }
    }

    EventLoop *NextThread()
    {
        if (_thread_count == 0)
        {
            return _base_loop;
        }
        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};

class Connection;
using ConnectionPtr = std::shared_ptr<Connection>;
/// @brief 连接关闭 正在连接 连接成功 关闭连接中
enum ConnectStatus
{
    DISCONNECTED,
    CONNECTING,
    CONNECTED,
    DISCONNECTING
};
class Connection : public std::enable_shared_from_this<Connection>
{

private:
    uint64_t _connect_id;          // 唯一id?
    int _socketfd;                 //
    Socket _socket;                // 监听套接字
    EventLoop *_loop;              // 事件监听和管理
    Channel _channel;              // 监听channel
    Buffer _in_buffer;             // 输入缓冲区
    Buffer _out_buffer;            // 输出缓冲区
    Any _context;                  // 上下文数据
    bool _enable_inactive_release; // 非活跃单位超时销毁标志位
    ConnectStatus _connect_status; // 连接状态标志位

    using ConnectedCallback = std::function<void(const ConnectionPtr &)>;
    using MessageCallback = std::function<void(const ConnectionPtr &, Buffer *)>;
    using CloseCallback = std::function<void(const ConnectionPtr &)>;
    using AnyEventCallback = std::function<void(const ConnectionPtr &)>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    CloseCallback _close_callback;
    AnyEventCallback _any_event_callback;
    CloseCallback _server_close_callback; // 移除组件内管理信息？？？

private:
    void HandleRead()
    {
        char buf[1024 * 10];
        // std::cout << __LINE__ << "  444" << std::endl;
        int ret = _socket.NoneBlockRecv(buf, 1024 * 10 - 1);
        if (ret <= 0) //<出错了 =0对端直接关了
        {
            if (ret == 0)
            {
                // std::cout << "ii" << std::endl;
            }
            // 在这里不能直接关了,读缓冲区的剩余数据可以直接调用回掉处理，
            // 但写剩余数据还得事件监控处理所以最终关闭的时间在写时间那里处理连带处理读缓冲区剩余数据
            //  读出错了处理读写数据缓存区的数据
            return ShutdownInLoop();
        }

        _in_buffer.Write(buf, ret);
        if (_in_buffer.ReadableSzie() > 0)
        {
            return _message_callback(shared_from_this(), &_in_buffer);
        }
    }
    void HandleWrite()
    {
        int ret = _socket.NoneBlockSend(_out_buffer.GetReadPostion(), _out_buffer.ReadableSzie());
        if (ret < 0)
        {

            // 出错时如果有读数据给它处理了没有就直接关了
            if (_in_buffer.ReadableSzie() > 0)
                _message_callback(shared_from_this(), &_in_buffer);
            return Release();
        }
        _out_buffer.MoveReadPostion(ret);
        if (_out_buffer.ReadableSzie() == 0)
        {
            // 无数据则关闭写监控
            _channel.DisableWrite();
            if (_connect_status == ConnectStatus::DISCONNECTING)
            {
                return Release();
            }
        }
    }
    void HandleClose()
    {
        // << __LINE__ << " handleclose" << std::endl;
        if (_in_buffer.ReadableSzie() > 0)
            _message_callback(shared_from_this(), &_in_buffer);
        return Release();
    }
    void HandleError()
    {
        return HandleClose();
    }
    void HandleEvent()
    {
        if (_enable_inactive_release)
            _loop->TimerRefresh(_connect_id);
        if (_any_event_callback)
            _any_event_callback(shared_from_this());
    }

    void EstablishedInLoop()
    {
        assert(_connect_status == ConnectStatus::CONNECTING);
        _connect_status = ConnectStatus::CONNECTED; ///////////////
        _channel.EnableRead();
        if (_connected_callback)
            _connected_callback(shared_from_this());
    }
    void ReleaseInLoop()
    {

        _connect_status = ConnectStatus::DISCONNECTED;
        // std::cout << "remove " << Fd() << std::endl;
        _channel.Remove(); // 移除事件监控
        // std::cout << "remove end" << std::endl;
        // std::cout << _connect_id << " close " << _socketfd << std::endl;
        _socket.Close(); // 移除套接字
        // std::cout << _connect_id << " closed " << _socketfd << std::endl;
        if (_loop->HasTimer(_connect_id)) // 当我已经开始释放时超时释放就要删除以防重复调用
        {
            DisablelInactiveReleasInLoop();
        }
        if (_close_callback)
            _close_callback(shared_from_this());
        if (_server_close_callback)
            _server_close_callback(shared_from_this());
        // std::cout<<__LINE__<<" tttt"<<std::endl;
    }
    /// @brief 将数据写到缓冲区通过事件监控将其发送具体 HandleWrite
    /// @param date
    /// @param n
    void SendInLoop(Buffer date)
    {
        if (_connect_status == ConnectStatus::DISCONNECTED) // 如果处于断连则不做操作
            return;
        _out_buffer.Write(date);
        if (_channel.isWriteable() == false)
            _channel.EnableWrite();
    }
    // 处理读写缓冲区剩余数据
    void ShutdownInLoop()
    {
        _connect_status = ConnectStatus::DISCONNECTING; // 设置当前状态为正在关闭中
        if (_in_buffer.ReadableSzie() > 0)              // 读区如有数据处理它
        {
            _message_callback(shared_from_this(), &_in_buffer);
        }
        if (_out_buffer.ReadableSzie() > 0) // 写区如果有数据则启动写事件
        {
            if (_channel.isWriteable() == false)
                _channel.EnableWrite();
        }
        if (_out_buffer.ReadableSzie() == 0) // 写区无数据读写缓冲区数据处理完毕直接释放
        {
            Release();
        }
    }
    void EnableInactiveReleaseInLoop(uint64_t sec)
    {
        _enable_inactive_release = true;
        if (_loop->HasTimer(_connect_id))
        {
            _loop->TimerRefresh(_connect_id);
        }
        else
        {
            _loop->TimerAdd(_connect_id, sec, std::bind(&Connection::ReleaseInLoop, this));
        }
    }
    void DisablelInactiveReleasInLoop()
    {
        _enable_inactive_release = false;
        _loop->TimerChannel(_connect_id); // 没添加过没关系不做任何操作
    }
    void UpgradeInLoop(const ConnectedCallback &conn,
                       const MessageCallback &mess,
                       const CloseCallback &close,
                       const AnyEventCallback &any,
                       const Any &context) // 切换协议？
    {
        _connected_callback = conn;
        _message_callback = mess;
        _close_callback = close;
        _any_event_callback = any;
        _context = context;
    }

public:
    // 以下接口为用户接口需要是对实际的调用及封装
    Connection(uint64_t connect_id, int socketfd, EventLoop *loop) : _connect_id(connect_id),
                                                                     _socketfd(socketfd),
                                                                     _socket(socketfd),
                                                                     _loop(loop),
                                                                     _channel(_socketfd, _loop),
                                                                     _enable_inactive_release(false),
                                                                     _connect_status(ConnectStatus::CONNECTING)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
        // 不可直接启动读事件还有一个EstablishedInLoop
    }
    ~Connection() { lg.logmessage(Debug, "addrse is%p", this); }
    int Fd() { return _socketfd; }
    int Id() { return _connect_id; }
    bool IsConnected() { return _connect_status == ConnectStatus::CONNECTED; }
    void SetContext(const Any &context) { _context = context; }
    Any *GetContext() { return &_context; }
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }  // 给用户使用
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }        // 给用户使用
    void SetCloseCallback(const CloseCallback &cb) { _close_callback = cb; }              // 给用户使用
    void SetAnyEventCallback(const AnyEventCallback &cb) { _any_event_callback = cb; }    // 给用户使用
    void SetServerCloseCallback(const CloseCallback &cb) { _server_close_callback = cb; } // 给上层封装tcpserevr用
    void Established() { _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this)); }
    void Send(const char *date, size_t n)
    {
        Buffer buf;
        buf.Write(date, n);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    void Release()
    {
        _loop->QueueInLOOP(std::bind(&Connection::ReleaseInLoop, this));
    }
    void EnableInactiveRelease(uint64_t sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    void DisablelInactiveReleas()
    {
        _loop->RunInLoop(std::bind(&Connection::DisablelInactiveReleasInLoop, this));
    }
    void Upgrade(const ConnectedCallback &conn,
                 const MessageCallback &mess,
                 const CloseCallback &close,
                 const AnyEventCallback &any,
                 const Any context) // 切换协议？
    {
        // 由于担心它在线程外执行进行排队时有数据但协议还未且所以必许立即执行
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, conn, mess, close, any, context));
    }
};

class Acceptor
{
    using AcceptorCallBack = std::function<void(int)>;

private:
    Socket _socket;
    EventLoop *_loop;
    Channel _channel;
    AcceptorCallBack _cb;

    int CreatServer(uint16_t port)
    {
        bool ret = _socket.CreatServer(port);
        assert(ret == true);
        return _socket.Fd();
    }

    void HandleRead()
    {
        int fd = _socket.Accept();
        if (fd < 0)
        {
            lg.logmessage(Error, "acept failure %s", strerror(errno));
            return;
        }
        lg.logmessage(Info, +"%d accept a new link %d", __LINE__, fd);
        if (_cb)
            _cb(fd);
    }

public:
    Acceptor(uint16_t port, EventLoop *loop) : _socket(CreatServer(port)), _loop(loop), _channel(_socket.Fd(), _loop)
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    void SetAcceptorCallBack(const AcceptorCallBack &cb)
    {
        _cb = cb;
        _channel.EnableRead();
    }
};
class TcpServer
{
private:
    using ConnectionPtr = std::shared_ptr<Connection>;
    using Func = std::function<void(void)>;
    using ConnectedCallback = std::function<void(const ConnectionPtr &)>;
    using MessageCallback = std::function<void(const ConnectionPtr &, Buffer *)>;
    using CloseCallback = std::function<void(const ConnectionPtr &)>;
    using AnyEventCallback = std::function<void(const ConnectionPtr &)>;
    uint64_t _next_id; // 一个增长的id不能用有符号的应为-5%3=-2
    EventLoop _base_loop;
    Acceptor _acceptor;
    LoopThreadPool _threads;                            // 线程分配
    std::unordered_map<uint64_t, ConnectionPtr> _conns; // 管理client连接
    bool _enable_inactive_release = false;
    int _time_out;

    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    CloseCallback _close_callback;
    AnyEventCallback _any_event_callback;
    CloseCallback _server_close_callback; // 移除组件内管理信息？？？

    void NewConnection(int fd)
    {
        _next_id++;
        std::cout << __LINE__ << "_next_id: " << _next_id << " acept a new " << fd << std::endl;
        ConnectionPtr conn_ptr(new Connection(_next_id, fd, _threads.NextThread())); //
        conn_ptr->SetMessageCallback(_message_callback);
        conn_ptr->SetCloseCallback(_close_callback);
        conn_ptr->SetConnectedCallback(_connected_callback);
        conn_ptr->SetServerCloseCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if (_enable_inactive_release)
            conn_ptr->EnableInactiveRelease(_time_out);
        conn_ptr->Established();
        _conns.insert(std::make_pair(_next_id, conn_ptr));
    }
    void RemoveConnectionInLoop(const ConnectionPtr &conn)
    {
        auto item = _conns.find(conn->Id());
        if (item != _conns.end())
        {
            _conns.erase(item);
        }
    }
    void RunAfterInLoop(const Func &task, int delay)
    {
        _next_id++;
        _base_loop.TimerAdd(_next_id, delay, task);
    }

public:
    TcpServer(uint16_t port, int thread_count = 1) : _next_id(0),
                                                     _acceptor(port, &_base_loop),
                                                     _threads(&_base_loop, thread_count),
                                                     _time_out(0),
                                                     _enable_inactive_release(false)
    {
        _acceptor.SetAcceptorCallBack(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1)); // 顺便启动读事件了
        //_threads.SetThreadCount(thread_count);
        _threads.Creat();
    }

    void SetThreadCount(int count) { _threads.SetThreadCount(count); } /// ？？后期可更改线程池大小
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetCloseCallback(const CloseCallback &cb) { _close_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _any_event_callback = cb; }

    void Start()
    {
        _base_loop.Start(); //
    }

    void RemoveConnection(const ConnectionPtr &conn)
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }
    void RunAfter(const Func &task, int delay) // 定时执行任务
    {
        _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }

    void EnableInactiveRelease(uint64_t time_out)
    {
        _time_out = time_out;
        _enable_inactive_release = true;
    }
    void DisableInactiveRelease()
    {
        _enable_inactive_release = false;
    }
};

class EchoServer
{
private:
    using ConnectionPtr = std::shared_ptr<Connection>;
    TcpServer _server;
    void OnConnected(const ConnectionPtr &connPtr)
    {
        lg.logmessage(Debug, "%d Connected %d", __LINE__, connPtr->Id());
    }
    void OnMessage(const ConnectionPtr &connPtr, Buffer *buf)
    {
        std::string mess = buf->ReadAsString(buf->ReadableSzie(), true);
        // std::cout<<mess<<std::endl;
        connPtr->Send(mess.c_str(), mess.size());
    }
    void OnClosed(const ConnectionPtr &connPtr)
    {
        lg.logmessage(Debug, "%d closed %d", __LINE__, connPtr->Id());
    }

public:
    EchoServer(uint16_t port) : _server(port, 2)
    {
        _server.SetMessageCallback(std::bind(&EchoServer::OnMessage, this, std::placeholders::_1, std::placeholders::_2));
        _server.SetConnectedCallback(std::bind(&EchoServer::OnConnected, this, std::placeholders::_1));
        _server.SetCloseCallback(std::bind(&EchoServer::OnClosed, this, std::placeholders::_1));
        _server.EnableInactiveRelease(5);
    }

    void Start()
    {
        _server.Start();
    }
};

void Channel::Remove() { _loop->RemoveEvent(this); }
/////它改变的是你要监控哪些事件
void Channel::Update() { _loop->UpdateEvent(this); }

void TimerWheel::AddInLoop(uint64_t id, uint32_t timeout, const TaskFunc &task)
{
    _loop->RunInLoop(std::bind(&TimerWheel::Add, this, id, timeout, task));
}

void TimerWheel::TimeRefreshInLoop(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimeRefresh, this, id));
}

void TimerWheel::CancleTaskInLoop(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::CancleTask, this, id));
}
